package com.example.repurchase.prediction.sink;

import com.example.repurchase.prediction.UserRepurchasePrediction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.sniff.Sniffer;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.transport.client.PreBuiltTransportClient;

import java.net.InetAddress;
import java.util.HashMap;
import java.util.Map;

public class MetricsToElasticsearch extends RichSinkFunction<UserRepurchasePrediction.UserRepurchaseScore> {

    private RestHighLevelClient client;
    private Sniffer sniffer;
    private final String hostname;
    private final int port;
    private final String index;

    public MetricsToElasticsearch(String hostname, int port, String index) {
        this.hostname = hostname;
        this.port = port;
        this.index = index;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        // 创建Elasticsearch客户端
        client = createClient(hostname, port);
    }

    @Override
    public void invoke(UserRepurchasePrediction.UserRepurchaseScore score, Context context) throws Exception {
        // 准备文档数据
        Map<String, Object> doc = new HashMap<>();
        doc.put("userId", score.getUserId());
        doc.put("timestamp", System.currentTimeMillis());
        doc.put("windowEndTime", score.getWindowEndTime());
        doc.put("purchaseFrequency", score.getPurchaseFrequency());
        doc.put("avgPurchaseAmount", score.getAvgPurchaseAmount());
        doc.put("repurchaseScore", score.getRepurchaseScore());

        // 创建索引请求
        IndexRequest request = new IndexRequest(index)
                .source(doc, XContentType.JSON);

        // 发送请求
        client.index(request, RequestOptions.DEFAULT);
    }

    @Override
    public void close() throws Exception {
        super.close();
        // 关闭客户端
        if (sniffer != null) {
            sniffer.close();
        }
        if (client != null) {
            client.close();
        }
    }

    private RestHighLevelClient createClient(String hostname, int port) {
        // 创建并配置Elasticsearch客户端
        // 注意：这里使用的是简化版本，实际生产环境中可能需要更复杂的配置
        return null; // 实际实现中需要根据Elasticsearch版本和客户端API进行配置
    }
}    